package org.apache.rocketmq.store;

import org.apache.commons.lang3.StringUtils;
import org.apache.rocketmq.common.*;
import org.apache.rocketmq.common.constant.LoggerName;
import org.apache.rocketmq.common.message.MessageConst;
import org.apache.rocketmq.common.message.MessageDecoder;
import org.apache.rocketmq.common.message.MessageExtBrokerInner;
import org.apache.rocketmq.common.sysflag.MessageSysFlag;
import org.apache.rocketmq.common.utils.QueueTypeUtils;
import org.apache.rocketmq.common.utils.ThreadUtils;
import org.apache.rocketmq.logging.org.slf4j.Logger;
import org.apache.rocketmq.logging.org.slf4j.LoggerFactory;
import org.apache.rocketmq.store.config.BrokerRole;
import org.apache.rocketmq.store.config.MessageStoreConfig;
import org.apache.rocketmq.store.config.StorePathConfigHelper;
import org.apache.rocketmq.store.hook.PutMessageHook;
import org.apache.rocketmq.store.hook.SendMessageBackHook;
import org.apache.rocketmq.store.index.IndexService;
import org.apache.rocketmq.store.logfile.MappedFile;
import org.apache.rocketmq.store.queue.ConsumeQueueInterface;
import org.apache.rocketmq.store.queue.ConsumeQueueStore;
import org.apache.rocketmq.store.queue.ConsumeQueueStoreInterface;
import org.apache.rocketmq.store.stats.BrokerStatsManager;
import org.rocksdb.RocksDBException;

import java.io.File;
import java.io.IOException;
import java.io.RandomAccessFile;
import java.nio.ByteBuffer;
import java.nio.channels.FileLock;
import java.nio.charset.StandardCharsets;
import java.util.*;
import java.util.concurrent.*;

/**
 * @课程描述:从零带你写框架系列中的课程，整个系列包含netty，xxl-job，rocketmq，nacos，sofajraft，spring，springboot，disruptor，编译器，虚拟机等等。
 * @author：陈清风扬，个人微信号：chenqingfengyangjj。
 * @date:2025/6/25
 * @方法描述：默认的消息存储引擎，这个消息存储就应就会把生产者客户端生产的消息提交给CommitLog对象，然后存储到CommitLog文件中
 * 可以说这个类的对象和CommitLog类的对象是rocketmq消息存储模块中最重要的两个功能对象
 */
public class DefaultMessageStore  implements MessageStore{


    protected static final Logger LOGGER = LoggerFactory.getLogger(LoggerName.STORE_LOGGER_NAME);

    protected static final Logger ERROR_LOG = LoggerFactory.getLogger(LoggerName.STORE_ERROR_LOGGER_NAME);

    //默认的消息存储引擎
    private final MessageStoreConfig messageStoreConfig;

    //存储引擎是否关闭的标志
    private volatile boolean shutdown = true;

    //当前存储引擎的运行状态
    protected final RunningFlags runningFlags = new RunningFlags();

    //定时任务执行器
    private final ScheduledExecutorService scheduledExecutorService;

    //broker信息统计管理器
    private final BrokerStatsManager brokerStatsManager;

    //消息到达监听器，这个监听器在第十三版本代码还用不上
    private final MessageArrivingListener messageArrivingListener;

    //broker配置信息
    private final BrokerConfig brokerConfig;

    //主题配置信息表
    private ConcurrentMap<String, TopicConfig> topicConfigTable;

    //执行队列清理任务的定时任务执行器
    private final ScheduledExecutorService scheduledCleanQueueExecutorService =
            ThreadUtils.newSingleThreadScheduledExecutor(new ThreadFactoryImpl("StoreCleanQueueScheduledThread"));

    //随机访问文件对象，用于创建文件锁
    private RandomAccessFile lockFile;

    //文件锁对象
    private FileLock lock;

    //存储引擎正常关闭标志
    boolean shutDownNormal = false;

    private SendMessageBackHook sendMessageBackHook;

    //消息延迟级别表
    private final ConcurrentSkipListMap<Integer /* level */, Long/* delay timeMillis */> delayLevelTable = new ConcurrentSkipListMap<>();

    //消息最大延迟级别
    private int maxDelayLevel;

    //状态机版本号
    private long stateMachineVersion = 0L;

    //系统时钟
    private final SystemClock systemClock = new SystemClock();

    private List<PutMessageHook> putMessageHookList = new ArrayList<>();

    //内存映射文件分配服务对象，这个成员变量就是用来创建MappedFile对象的
    private final AllocateMappedFileService allocateMappedFileService;

    //把消息存储到CommitLog文件中的核心组件
    protected final CommitLog commitLog;

    //拉取消息的最大大小
    private final static int MAX_PULL_MSG_SIZE = 128 * 1024 * 1024;

    //消息存储统计服务组件
    private final StoreStatsService storeStatsService;

    //消息临时缓冲区，其实就是对外缓冲区
    private final TransientStorePool transientStorePool;

    //消息存储检查点
    private StoreCheckpoint storeCheckpoint;

    private volatile long masterFlushedOffset = -1L;

    private volatile long brokerInitMaxOffset = -1L;

    //ConsumeQueue消息队列存储引擎
    protected final ConsumeQueueStoreInterface consumeQueueStore;

    //消息重放服务组件
    private ReputMessageService reputMessageService;

    //消费队列数据刷新服务组件
    private final FlushConsumeQueueService flushConsumeQueueService;

    //是否批量通知消息到达的标志
    protected boolean notifyMessageArriveInBatch = false;

    //存放消息分发器的集合
    private final LinkedList<CommitLogDispatcher> dispatcherList;

    //CommitLog过期文件清理组件
    //protected final CleanCommitLogService cleanCommitLogService;

    //ConsumeQueue过期文件清理组件
    //private final CleanConsumeQueueService cleanConsumeQueueService;

    //索引文件服务组件
    protected final IndexService indexService;

    //构造方法
    public DefaultMessageStore(final MessageStoreConfig messageStoreConfig, final BrokerStatsManager brokerStatsManager,
                               final MessageArrivingListener messageArrivingListener, final BrokerConfig brokerConfig, final ConcurrentMap<String, TopicConfig> topicConfigTable) throws IOException {

        this.messageArrivingListener = messageArrivingListener;
        this.brokerConfig = brokerConfig;
        this.messageStoreConfig = messageStoreConfig;
        this.brokerStatsManager = brokerStatsManager;
        this.topicConfigTable = topicConfigTable;
        this.allocateMappedFileService = new AllocateMappedFileService(this);
        this.commitLog = new CommitLog(this);
        this.consumeQueueStore = createConsumeQueueStore();
        this.flushConsumeQueueService = createFlushConsumeQueueService();
        this.storeStatsService = new StoreStatsService(getBrokerIdentity());
        this.indexService = new IndexService(this);
        this.reputMessageService =new ReputMessageService();
        this.transientStorePool = new TransientStorePool(messageStoreConfig.getTransientStorePoolSize(), messageStoreConfig.getMappedFileSizeCommitLog());
        this.scheduledExecutorService = ThreadUtils.newSingleThreadScheduledExecutor(new ThreadFactoryImpl("StoreScheduledThread", getBrokerIdentity()));
        this.dispatcherList = new LinkedList<>();
        this.dispatcherList.addLast(new CommitLogDispatcherBuildConsumeQueue());
        this.dispatcherList.addLast(new CommitLogDispatcherBuildIndex());
        //if (messageStoreConfig.isEnableCompaction()) {
        //this.compactionStore = new CompactionStore(this);
        //this.compactionService = new CompactionService(commitLog, this, compactionStore);
        //this.dispatcherList.addLast(new CommitLogDispatcherCompaction(compactionService));
        //}
        //创建文件锁所在的文件
        File file = new File(StorePathConfigHelper.getLockFile(messageStoreConfig.getStorePathRootDir()));
        //确保文件所在目录存在
        UtilAll.ensureDirOK(file.getParent());
        //确保物理存储路径存在，其实就是验证存储消息的CommitLog文件的路径
        UtilAll.ensureDirOK(getStorePathPhysic());
        //确保逻辑存储路径存在，其实就是验证存储消息的ConsumeQueue文件路径
        UtilAll.ensureDirOK(getStorePathLogic());
        //打开文件锁对应的文件
        lockFile = new RandomAccessFile(file, "rw");
        //解析延迟级别设置
        parseDelayLevel();
    }


    public ConsumeQueueStoreInterface createConsumeQueueStore() {
        return new ConsumeQueueStore(this);
    }

    public FlushConsumeQueueService createFlushConsumeQueueService() {
        return new FlushConsumeQueueService();
    }

    /**
     * @课程描述:从零带你写框架系列中的课程，整个系列包含netty，xxl-job，rocketmq，nacos，sofajraft，spring，springboot，disruptor，编译器，虚拟机等等。
     * @author：陈清风扬，个人微信号：chenqingfengyangjj。
     * @方法描述：消息引擎初始化加载数据的方法，该方法在第十三版本代码中并没有实现，因为我们连存储数据的完整功能都没有实现呢
     * 本地根本就没有数据，怎么加载呢，所以我对这个方法做了大量简化，大家简单看看就行
     */
    @Override
    public boolean load() {
        //定义默认的初始化结果标志
        boolean result = true;
        try {
            //检查是否存在临时文件，如果这个临时文件存在，意味着上一次程序关闭时出现了意外，不是正常关闭
            //那接下来就要执行数据恢复操作
            //如果程序是正常关闭，那么这个临时文件就会不存在，因为程序关闭前会把这个文件删除了
            boolean lastExitOK = !this.isTempFileExist();
            //记录程序上次关闭时的状态和临时文件的存储路径
            LOGGER.info("last shutdown {}, store path root dir: {}",
                    lastExitOK ? "normally" : "abnormally", messageStoreConfig.getStorePathRootDir());
            //加载消息数据的操作，也就是加载内存映射文件的操作
            result = this.commitLog.load();
            //加载消费队列内存映射文件
            result = result && this.consumeQueueStore.load();
            if (result) {
                //创建并加载消息存储检查点文件信息，this.messageStoreConfig.getStorePathRootDir()得到的就是该文件的存储路径
                this.storeCheckpoint = new StoreCheckpoint(StorePathConfigHelper.getStoreCheckpoint(this.messageStoreConfig.getStorePathRootDir()));
                //设置broker主节点刷新消息偏移量
                this.masterFlushedOffset = this.storeCheckpoint.getMasterFlushedOffset();
                //设置当前broker节点已确认的消息的物理偏移量
                setConfirmOffset(this.storeCheckpoint.getConfirmPhyOffset());
                //如果以上操作都成功，那就在这里根据临时文件是否存在执行是否正常恢复数据的操作
                this.recover(lastExitOK);
            }
        } catch (Exception e) {
            LOGGER.error("load exception", e);
            result = false;
        }
        return result;
    }


    //数据恢复的方法，该方法暂时不做实现
    private void recover(boolean lastExitOK) {
    }


    /**
     * @课程描述:从零带你写框架系列中的课程，整个系列包含netty，xxl-job，rocketmq，nacos，sofajraft，spring，springboot，disruptor，编译器，虚拟机等等。
     * @author：陈清风扬，个人微信号：chenqingfengyangjj。
     * @方法描述：启动消息存储引擎的方法，该方法在第十三版本代码中做了大量简化
     */
    @Override
    public void start() throws Exception {

        //判断是否启用了临时缓冲池功能，启用了则初始化临时缓冲池对象
        if (this.isTransientStorePoolEnable()) {
            this.transientStorePool.init();
        }
        //启动内存映射分配服务
        this.allocateMappedFileService.start();

        //尝试获取文件锁，确保不会有多个实例同时运行
        lock = lockFile.getChannel().tryLock(0, 1, false);
        if (lock == null || lock.isShared() || !lock.isValid()) {
            throw new RuntimeException("Lock failed, MQ already started");
        }

        //在文件锁对应的文件中写入"lock"，并强制刷新到磁盘，确保文件锁的持有
        lockFile.getChannel().write(ByteBuffer.wrap("lock".getBytes(StandardCharsets.UTF_8)));
        lockFile.getChannel().force(true);
        //设置重放消息服务的起始偏移量
        this.reputMessageService.setReputFromOffset(this.commitLog.getConfirmOffset());
        //启动消息重放服务
        this.reputMessageService.start();
        //启动消费队列数据刷新服务
        this.flushConsumeQueueService.start();
        //启动消息提交服务
        this.commitLog.start();
        this.consumeQueueStore.start();
        this.storeStatsService.start();
        //在这里创建了临时文件，这个临时文件会检测程序的退出状态，如果正常退出，那这个文件就会被删除
        //如果异常退出，则这个文件会被保留，用于下次启动时判断程序是否正常退出
        this.createTempFile();
        //初始化定时任务
        this.addScheduleTask();
        //设置关闭标志为false，表示服务正在运行
        this.shutdown = false;
    }


    /**
     * @课程描述:从零带你写框架系列中的课程，整个系列包含netty，xxl-job，rocketmq，nacos，sofajraft，spring，springboot，disruptor，编译器，虚拟机等等。
     * @author：陈清风扬，个人微信号：chenqingfengyangjj。
     * @方法描述：初始化定时任务的方法，该方法在第十三版本代码中也做了大量简化
     */
    private void addScheduleTask() {
        this.scheduledExecutorService.scheduleAtFixedRate(new AbstractBrokerRunnable(this.getBrokerIdentity()) {
            @Override
            public void run0() {
                //执行清理过期文件的操作，根据过期策略删除已近过期的CommitLog文件
                DefaultMessageStore.this.cleanFilesPeriodically();
            }
        }, 1000 * 60, this.messageStoreConfig.getCleanResourceInterval(), TimeUnit.MILLISECONDS);

        //内存映射文件定期执行自检操作
        this.scheduledExecutorService.scheduleAtFixedRate(new AbstractBrokerRunnable(this.getBrokerIdentity()) {
            @Override
            public void run0() {
                DefaultMessageStore.this.checkSelf();
            }
        }, 1, 10, TimeUnit.MINUTES);

        //定期刷新存储检查点数据到本地硬盘文件
        this.scheduledExecutorService.scheduleAtFixedRate(
                new AbstractBrokerRunnable(this.getBrokerIdentity()) {
                    @Override
                    public void run0() {
                        DefaultMessageStore.this.storeCheckpoint.flush();
                    }
                },
                1,
                1,
                TimeUnit.SECONDS
        );

        //定期清理过期消费队列文件任务
        this.scheduledCleanQueueExecutorService.scheduleAtFixedRate(
                new Runnable() {
                    @Override
                    public void run() {
                        DefaultMessageStore.this.cleanQueueFilesPeriodically();
                    }
                },
                1000 * 60,
                this.messageStoreConfig.getCleanResourceInterval(),
                TimeUnit.MILLISECONDS
        );
    }


    /**
     * @课程描述:从零带你写框架系列中的课程，整个系列包含netty，xxl-job，rocketmq，nacos，sofajraft，spring，springboot，disruptor，编译器，虚拟机等等。
     * @author：陈清风扬，个人微信号：chenqingfengyangjj。
     * @方法描述：清理过期CommitLog文件数据的方法
     */
    private void cleanFilesPeriodically() {
        //this.cleanCommitLogService.run();
    }


    /**
     * @课程描述:从零带你写框架系列中的课程，整个系列包含netty，xxl-job，rocketmq，nacos，sofajraft，spring，springboot，disruptor，编译器，虚拟机等等。
     * @author：陈清风扬，个人微信号：chenqingfengyangjj。
     * @方法描述：定期清理过期的消费队列文件的方法
     */
    private void cleanQueueFilesPeriodically() {
        //this.correctLogicOffsetService.run();
        //this.cleanConsumeQueueService.run();
    }

    /**
     * @课程描述:从零带你写框架系列中的课程，整个系列包含netty，xxl-job，rocketmq，nacos，sofajraft，spring，springboot，disruptor，编译器，虚拟机等等。
     * @author：陈清风扬，个人微信号：chenqingfengyangjj。
     * @方法描述：内存映射文件定期执行自检的惭怍
     */
    private void checkSelf() {
        //检查CommitLog文件的完整性
        this.commitLog.checkSelf();
        //检查所有ConsumeQueue的完整性
        this.consumeQueueStore.checkSelf();
    }

    /**
     * @课程描述:从零带你写框架系列中的课程，整个系列包含netty，xxl-job，rocketmq，nacos，sofajraft，spring，springboot，disruptor，编译器，虚拟机等等。
     * @author：陈清风扬，个人微信号：chenqingfengyangjj。
     * @方法描述：判断临时文件是否存在的方法，也就是判断程序上一次是否正常关闭
     */
    private boolean isTempFileExist() {
        String fileName = StorePathConfigHelper.getAbortFile(this.messageStoreConfig.getStorePathRootDir());
        File file = new File(fileName);
        return file.exists();
    }


    /**
     * @课程描述:从零带你写框架系列中的课程，整个系列包含netty，xxl-job，rocketmq，nacos，sofajraft，spring，springboot，disruptor，编译器，虚拟机等等。
     * @author：陈清风扬，个人微信号：chenqingfengyangjj。
     * @方法描述：解析消息延迟级别的方法，其实就是把存储引擎配置信息中的延迟级别解析到delayLevelTable成员变量中的方法，大家简单看看就行
     */
    public boolean parseDelayLevel() {
        HashMap<String, Long> timeUnitTable = new HashMap<>();
        timeUnitTable.put("s", 1000L);
        timeUnitTable.put("m", 1000L * 60);
        timeUnitTable.put("h", 1000L * 60 * 60);
        timeUnitTable.put("d", 1000L * 60 * 60 * 24);
        String levelString = messageStoreConfig.getMessageDelayLevel();
        try {
            String[] levelArray = levelString.split(" ");
            for (int i = 0; i < levelArray.length; i++) {
                String value = levelArray[i];
                String ch = value.substring(value.length() - 1);
                Long tu = timeUnitTable.get(ch);
                int level = i + 1;
                if (level > this.maxDelayLevel) {
                    this.maxDelayLevel = level;
                }
                long num = Long.parseLong(value.substring(0, value.length() - 1));
                long delayTimeMillis = tu * num;
                this.delayLevelTable.put(level, delayTimeMillis);
            }
        } catch (Exception e) {
            LOGGER.error("parse message delay level failed. messageDelayLevel = {}", levelString, e);
            return false;
        }
        return true;
    }

    //得到消息存储引擎的运行状态
    @Override
    public RunningFlags getRunningFlags() {
        return runningFlags;
    }



    /**
     * @课程描述:从零带你写框架系列中的课程，整个系列包含netty，xxl-job，rocketmq，nacos，sofajraft，spring，springboot，disruptor，编译器，虚拟机等等。
     * @author：陈清风扬，个人微信号：chenqingfengyangjj。
     * @方法描述：同步存储消息的方法
     */
    @Override
    public PutMessageResult putMessage(MessageExtBrokerInner msg) {
        //在这里同步等待消息存储完成，asyncPutMessage就是异步存储消息的操作
        return waitForPutResult(asyncPutMessage(msg));
    }


    /**
     * @课程描述:从零带你写框架系列中的课程，整个系列包含netty，xxl-job，rocketmq，nacos，sofajraft，spring，springboot，disruptor，编译器，虚拟机等等。
     * @author：陈清风扬，个人微信号：chenqingfengyangjj。
     * @方法描述：同步等待消息存储结果的方法
     */
    private PutMessageResult waitForPutResult(CompletableFuture<PutMessageResult> putMessageResultFuture) {
        try {
            //计算等待超时时间，根据消息同步刷新超时、从节点同步超时中的最大值并加上5秒得到等待超时的时间
            int putMessageTimeout = Math.max(this.messageStoreConfig.getSyncFlushTimeout(), this.messageStoreConfig.getSlaveTimeout()) + 5000;
            //等待处理结果
            return putMessageResultFuture.get(putMessageTimeout, TimeUnit.MILLISECONDS);
        } catch (ExecutionException | InterruptedException e) {
            return new PutMessageResult(PutMessageStatus.UNKNOWN_ERROR, null);
        } catch (TimeoutException e) {
            LOGGER.error("usually it will never timeout, putMessageTimeout is much bigger than slaveTimeout and "
                    + "flushTimeout so the result can be got anyway, but in some situations timeout will happen like full gc "
                    + "process hangs or other unexpected situations.");
            return new PutMessageResult(PutMessageStatus.UNKNOWN_ERROR, null);
        }
    }


    /**
     * @课程描述:从零带你写框架系列中的课程，整个系列包含netty，xxl-job，rocketmq，nacos，sofajraft，spring，springboot，disruptor，编译器，虚拟机等等。
     * @author：陈清风扬，个人微信号：chenqingfengyangjj。
     * @方法描述：异步存储消息的方法，把单挑消息提交给commitLog对象存储到本地
     */
    @Override
    public CompletableFuture<PutMessageResult> asyncPutMessage(MessageExtBrokerInner msg) {
        //遍历消息存储前的回调方法
        for (PutMessageHook putMessageHook : putMessageHookList) {
            //执行消息存储前的每一个回调方法
            PutMessageResult handleResult = putMessageHook.executeBeforePutMessage(msg);
            //如果回调方法返回了非空的处理结果，则直接返回该结果的完成状态
            if (handleResult != null) {
                return CompletableFuture.completedFuture(handleResult);
            }
        }

        //检查消息属性中是否设置了内部批量消息的标志，但消息系统标志中未设置批量标志
        if (msg.getProperties().containsKey(MessageConst.PROPERTY_INNER_NUM)
                && !MessageSysFlag.check(msg.getSysFlag(), MessageSysFlag.INNER_BATCH_FLAG)) {
            //如果存在这种情况，则记录警告日志
            LOGGER.warn("[BUG]The message had property {} but is not an inner batch", MessageConst.PROPERTY_INNER_NUM);
            //返回消息非法的错误状态
            return CompletableFuture.completedFuture(new PutMessageResult(PutMessageStatus.MESSAGE_ILLEGAL, null));
        }

        //检查消息是否为内部批量消息
        if (MessageSysFlag.check(msg.getSysFlag(), MessageSysFlag.INNER_BATCH_FLAG)) {
            //如果是内部批量消息，则获取消息主题的配置信息
            Optional<TopicConfig> topicConfig = this.getTopicConfig(msg.getTopic());
            //检查主题配置信息中是否设置了批量消费队列类型
            if (!QueueTypeUtils.isBatchCq(topicConfig)) {
                //如果不是，记录错误日志
                LOGGER.error("[BUG]The message is an inner batch but cq type is not batch cq");
                //返回消息非法的错误状态
                return CompletableFuture.completedFuture(new PutMessageResult(PutMessageStatus.MESSAGE_ILLEGAL, null));
            }
        }

        //记录开始存储消息的时间
        long beginTime = this.getSystemClock().now();
        //在这里把消息提交给commitLog对象异步存储
        CompletableFuture<PutMessageResult> putResultFuture = this.commitLog.asyncPutMessage(msg);
        //向future对象注册回调方法，处理消息存储结果
        putResultFuture.thenAccept(result -> {
            //计算处理消息所花费的时间
            long elapsedTime = this.getSystemClock().now() - beginTime;
            //如果处理时间超过500毫秒，记录警告日志
            if (elapsedTime > 500) {
                LOGGER.warn("DefaultMessageStore#putMessage: CommitLog#putMessage cost {}ms, topic={}, bodyLength={}",
                        elapsedTime, msg.getTopic(), msg.getBody().length);
            }
            //更新存储统计服务中的最大消息处理时间
            this.storeStatsService.setPutMessageEntireTimeMax(elapsedTime);
            //如果处理结果为空或结果状态不是OK，增加失败次数的统计
            if (null == result || !result.isOk()) {
                this.storeStatsService.getPutMessageFailedTimes().add(1);
            }
        });
        return putResultFuture;
    }


    //得到commitLog文件存储路径
    public String getStorePathPhysic() {
        return DefaultMessageStore.this.getMessageStoreConfig().getStorePathCommitLog();
    }


    //得到ConsumeQueue文件存储路径
    public String getStorePathLogic() {
        return StorePathConfigHelper.getStorePathConsumeQueue(this.messageStoreConfig.getStorePathRootDir());
    }


    /**
     * @课程描述:从零带你写框架系列中的课程，整个系列包含netty，xxl-job，rocketmq，nacos，sofajraft，spring，springboot，disruptor，编译器，虚拟机等等。
     * @author：陈清风扬，个人微信号：chenqingfengyangjj。
     * @方法描述：创建临时文件的方法，如果程序正常退出，这个临时文件会被删除，如果程序异常退出，这个临时文件会被保留，下次启动的时候会根据这个临时文件判断程序是否正常退出，是否需要执行数据恢复操作
     */
    private void createTempFile() throws IOException {
        String fileName = StorePathConfigHelper.getAbortFile(this.messageStoreConfig.getStorePathRootDir());
        File file = new File(fileName);
        UtilAll.ensureDirOK(file.getParent());
        boolean result = file.createNewFile();
        LOGGER.info(fileName + (result ? " create OK" : " already exists"));
        //将当前进程PID写入文件内容
        MixAll.string2File(Long.toString(MixAll.getPID()), file.getAbsolutePath());
    }


    /**
     * @课程描述:从零带你写框架系列中的课程，整个系列包含netty，xxl-job，rocketmq，nacos，sofajraft，spring，springboot，disruptor，编译器，虚拟机等等。
     * @author：陈清风扬，个人微信号：chenqingfengyangjj。
     * @方法描述：得到当前broker节点的身份标识
     */
    public BrokerIdentity getBrokerIdentity() {
        if (messageStoreConfig.isEnableDLegerCommitLog()) {
            return new BrokerIdentity(
                    brokerConfig.getBrokerClusterName(), brokerConfig.getBrokerName(),
                    Integer.parseInt(messageStoreConfig.getdLegerSelfId().substring(1)), brokerConfig.isInBrokerContainer());
        } else {
            return new BrokerIdentity(
                    brokerConfig.getBrokerClusterName(), brokerConfig.getBrokerName(),
                    brokerConfig.getBrokerId(), brokerConfig.isInBrokerContainer());
        }
    }

    /**
     * @课程描述:从零带你写框架系列中的课程，整个系列包含netty，xxl-job，rocketmq，nacos，sofajraft，spring，springboot，disruptor，编译器，虚拟机等等。
     * @author：陈清风扬，个人微信号：chenqingfengyangjj。
     * @方法描述：程序终止工作的方法
     */
    @Override
    public void shutdown() {
        if (!this.shutdown) {
            this.shutdown = true;

            this.scheduledExecutorService.shutdown();
            this.scheduledCleanQueueExecutorService.shutdown();

            try {
                this.scheduledExecutorService.awaitTermination(3, TimeUnit.SECONDS);
                this.scheduledCleanQueueExecutorService.awaitTermination(3, TimeUnit.SECONDS);
                Thread.sleep(1000 * 3);
            } catch (InterruptedException e) {
                LOGGER.error("shutdown Exception, ", e);
            }
            this.storeStatsService.shutdown();
            this.commitLog.shutdown();
            this.reputMessageService.shutdown();
            this.consumeQueueStore.shutdown();
            this.flushConsumeQueueService.shutdown();
            this.allocateMappedFileService.shutdown();
            this.storeCheckpoint.flush();
            this.storeCheckpoint.shutdown();
            //在这里会判断一下消息存储引擎是否可写，并且未分发消息的总字节数等于0，如果消息已经分发完毕了就可以正常关闭服务
            //现在大家可能还不理解消息分发完毕指的是什么，没关系，再等两个版本代码大家就全清楚了
            if (this.runningFlags.isWriteable() && dispatchBehindBytes() == 0) {
                //在这里把临时文件删除
                this.deleteFile(StorePathConfigHelper.getAbortFile(this.messageStoreConfig.getStorePathRootDir()));
                //设置服务正常关闭为true
                shutDownNormal = true;
            } else {
                LOGGER.warn("the store may be wrong, so shutdown abnormally, and keep abort file.");
            }
        }
        this.transientStorePool.destroy();
        if (lockFile != null && lock != null) {
            try {
                lock.release();
                lockFile.close();
            } catch (IOException e) {
            }
        }
    }


    /**
     * @课程描述:从零带你写框架系列中的课程，整个系列包含netty，xxl-job，rocketmq，nacos，sofajraft，spring，springboot，disruptor，编译器，虚拟机等等。
     * @author：陈清风扬，个人微信号：chenqingfengyangjj。
     * @方法描述：删除指定文件的方法
     */
    private void deleteFile(final String fileName) {
        File file = new File(fileName);
        boolean result = file.delete();
        LOGGER.info(fileName + (result ? " delete OK" : " delete Failed"));
    }


    /**
     * @课程描述:从零带你写框架系列中的课程，整个系列包含netty，xxl-job，rocketmq，nacos，sofajraft，spring，springboot，disruptor，编译器，虚拟机等等。
     * @author：陈清风扬，个人微信号：chenqingfengyangjj。
     * @方法描述：得到未分发消息字节总数的方法，该方法在第十四版本代码中还是伪实现
     */
    @Override
    public long dispatchBehindBytes() {
        //return this.reputMessageService.behind();
        return 0;
    }


    //销毁当前存储引擎的方法
    @Override
    public void destroy() {
        //销毁消费队列存储引擎
        this.consumeQueueStore.destroy();
        //销毁消息提交组件
        this.commitLog.destroy();
        //销毁索引服务组件
        //this.indexService.destroy();
        //删除中断文件，这是一个用于标记异常中断的文件
        //this.deleteFile(StorePathConfigHelper.getAbortFile(this.messageStoreConfig.getStorePathRootDir()));
        //删除存储检查点文件，这是用于记录消息存储关键点信息的文件
        this.deleteFile(StorePathConfigHelper.getStoreCheckpoint(this.messageStoreConfig.getStorePathRootDir()));
    }

    @Override
    public long getStateMachineVersion() {
        return stateMachineVersion;
    }

    public void setStateMachineVersion(long stateMachineVersion) {
        this.stateMachineVersion = stateMachineVersion;
    }


    public BrokerStatsManager getBrokerStatsManager() {
        return brokerStatsManager;
    }


    /**
     * @课程描述:从零带你写框架系列中的课程，整个系列包含netty，xxl-job，rocketmq，nacos，sofajraft，spring，springboot，disruptor，编译器，虚拟机等等。
     * @author：陈清风扬，个人微信号：chenqingfengyangjj。
     * @方法描述：判断操作系统页缓存是否繁忙的方法
     */
    @Override
    public boolean isOSPageCacheBusy() {
        //获取提交消息时加锁的开始时间
        long begin = this.getCommitLog().getBeginTimeInLock();
        //计算从开始加锁到现在的时间差
        long diff = this.systemClock.now() - begin;
        //如果时间差小于10000000并且大于配置的页面缓存繁忙超时时间，则认为页面缓存繁忙
        return diff < 10000000 && diff > this.messageStoreConfig.getOsPageCacheBusyTimeOutMills();
    }


    //得到系统当前时间的方法
    @Override
    public long now() {
        return this.systemClock.now();
    }


    public BrokerConfig getBrokerConfig() {
        return brokerConfig;
    }

    @Override
    public List<PutMessageHook> getPutMessageHookList() {
        return putMessageHookList;
    }


    @Override
    public void setSendMessageBackHook(SendMessageBackHook sendMessageBackHook) {
        this.sendMessageBackHook = sendMessageBackHook;
    }


    @Override
    public SendMessageBackHook getSendMessageBackHook() {
        return sendMessageBackHook;
    }


    @Override
    public boolean isShutdown() {
        return shutdown;
    }


    //临时缓冲池资源是否承租的方法
    @Override
    public boolean isTransientStorePoolDeficient() {
        return remainTransientStoreBufferNumbs() == 0;
    }


    //获取剩余可用的临时存储缓冲区数量的方法
    public int remainTransientStoreBufferNumbs() {
        //检查是否启用临时缓冲池功能
        if (this.isTransientStorePoolEnable()) {
            //返回缓冲池中剩余可用缓冲区数量
            return this.transientStorePool.availableBufferNums();
        }
        //未启用时返回理论最大值，表示没有限制
        return Integer.MAX_VALUE;
    }


    //得到broker主节点消息刷新偏移量的方法
    @Override
    public long getMasterFlushedOffset() {
        return this.masterFlushedOffset;
    }

    //设置broker主节点消息刷新偏移量的方法
    @Override
    public void setMasterFlushedOffset(long masterFlushedOffset) {
        this.masterFlushedOffset = masterFlushedOffset;
        //这里的操作就是把消息刷新偏移量写入到存储引擎检查点文件中
        this.storeCheckpoint.setMasterFlushedOffset(masterFlushedOffset);
    }


    /**
     * @课程描述:从零带你写框架系列中的课程，整个系列包含netty，xxl-job，rocketmq，nacos，sofajraft，spring，springboot，disruptor，编译器，虚拟机等等。
     * @author：陈清风扬，个人微信号：chenqingfengyangjj。
     * @方法描述：为要存储的消息分配消息队列偏移量的方法，这个方法就会把本条消息在ConsumeQueue目标队列中的偏移量明确了
     */
    @Override
    public void assignOffset(MessageExtBrokerInner msg) throws RocksDBException {
        //得到消息的具体类型
        final int tranType = MessageSysFlag.getTransactionValue(msg.getSysFlag());
        //如果是非事务消息或是事务提交消息
        if (tranType == MessageSysFlag.TRANSACTION_NOT_TYPE || tranType == MessageSysFlag.TRANSACTION_COMMIT_TYPE) {
            //那就在这里给消息分配消息偏移量
            this.consumeQueueStore.assignQueueOffset(msg);
        }
    }


    /**
     * @课程描述:从零带你写框架系列中的课程，整个系列包含netty，xxl-job，rocketmq，nacos，sofajraft，spring，springboot，disruptor，编译器，虚拟机等等。
     * @author：陈清风扬，个人微信号：chenqingfengyangjj。
     * @方法描述：增加ConsumeQueue消费队列偏移量的方法，其实就是记录一下这个消息队列中又存储了多少条消息
     */
    @Override
    public void increaseOffset(MessageExtBrokerInner msg, short messageNum) {
        final int tranType = MessageSysFlag.getTransactionValue(msg.getSysFlag());
        if (tranType == MessageSysFlag.TRANSACTION_NOT_TYPE || tranType == MessageSysFlag.TRANSACTION_COMMIT_TYPE) {
            this.consumeQueueStore.increaseQueueOffset(msg, messageNum);
        }
    }


    /**
     * @课程描述:从零带你写框架系列中的课程，整个系列包含netty，xxl-job，rocketmq，nacos，sofajraft，spring，springboot，disruptor，编译器，虚拟机等等。
     * @author：陈清风扬，个人微信号：chenqingfengyangjj。
     * @方法描述：查找指定主题和队列ID的消费队列的方法
     */
    @Override
    public ConsumeQueueInterface findConsumeQueue(String topic, int queueId) {
        return this.consumeQueueStore.findOrCreateConsumeQueue(topic, queueId);
    }


    /**
     * @课程描述:从零带你写框架系列中的课程，整个系列包含netty，xxl-job，rocketmq，nacos，sofajraft，spring，springboot，disruptor，编译器，虚拟机等等。
     * @author：陈清风扬，个人微信号：chenqingfengyangjj。
     * @方法描述：延迟解锁内存映射文件的方法
     */
    @Override
    public void unlockMappedFile(final MappedFile mappedFile) {
        this.scheduledExecutorService.schedule(new Runnable() {
            @Override
            public void run() {
                mappedFile.munlock();
            }
        }, 6, TimeUnit.SECONDS);
    }


    @Override
    public TransientStorePool getTransientStorePool() {
        return transientStorePool;
    }

    @Override
    public AllocateMappedFileService getAllocateMappedFileService() {
        return allocateMappedFileService;
    }


    /**
     * @课程描述:从零带你写框架系列中的课程，整个系列包含netty，xxl-job，rocketmq，nacos，sofajraft，spring，springboot，disruptor，编译器，虚拟机等等。
     * @author：陈清风扬，个人微信号：chenqingfengyangjj。
     * @方法描述：设置消息确认偏移量的方法，在非集群模式下，消息确认偏移量其实就是操作系统内存中最大的消息偏移量
     * 注意，这个并不是刷新到硬盘的消息偏移量
     */
    @Override
    public void setConfirmOffset(long phyOffset) {
        this.commitLog.setConfirmOffset(phyOffset);
    }


    @Override
    public StoreCheckpoint getStoreCheckpoint() {
        return storeCheckpoint;
    }


    /**
     * @课程描述:从零带你写框架系列中的课程，整个系列包含netty，xxl-job，rocketmq，nacos，sofajraft，spring，springboot，disruptor，编译器，虚拟机等等。
     * @author：陈清风扬，个人微信号：chenqingfengyangjj。
     * @方法描述：判断消息是否为冷数据的方法
     */
    public boolean checkInColdAreaByCommitOffset(long offsetPy, long maxOffsetPy) {
        //计算内存热点数据区域容量，就是总物理内存 × 配置的内存热点数据比率
        long memory = (long)(StoreUtil.TOTAL_PHYSICAL_MEMORY_SIZE * (this.messageStoreConfig.getAccessMessageInMemoryHotRatio() / 100.0));
        //判断消息区间长度是否超过内存热点区域容量，超过则是冷数据
        return (maxOffsetPy - offsetPy) > memory;
    }

    @Override
    public void onCommitLogAppend(MessageExtBrokerInner msg, AppendMessageResult result, MappedFile commitLogFile) {
        //在源码中就是空实现
    }



    public boolean isTransientStorePoolEnable() {
        return this.messageStoreConfig.isTransientStorePoolEnable() &&
                (this.brokerConfig.isEnableControllerMode() || this.messageStoreConfig.getBrokerRole() != BrokerRole.SLAVE);
    }


    @Override
    public MessageStoreConfig getMessageStoreConfig() {
        return messageStoreConfig;
    }


    public ConcurrentMap<String, TopicConfig> getTopicConfigs() {
        return this.topicConfigTable;
    }


    public Optional<TopicConfig> getTopicConfig(String topic) {
        if (this.topicConfigTable == null) {
            return Optional.empty();
        }
        return Optional.ofNullable(this.topicConfigTable.get(topic));
    }


    public SystemClock getSystemClock() {
        return systemClock;
    }


    @Override
    public CommitLog getCommitLog() {
        return commitLog;
    }

    @Override
    public StoreStatsService getStoreStatsService() {
        return storeStatsService;
    }


    /**
     * @课程描述:从零带你写框架系列中的课程，整个系列包含netty，xxl-job，rocketmq，nacos，sofajraft，spring，springboot，disruptor，编译器，虚拟机等等。
     * @author：陈清风扬，个人微信号：chenqingfengyangjj。
     * @方法描述：得到消费队列的内存映射文件表
     */
    public ConcurrentMap<String, ConcurrentMap<Integer, ConsumeQueueInterface>> getConsumeQueueTable() {
        return consumeQueueStore.getConsumeQueueTable();
    }



    /**
     * @课程描述:从零带你写框架系列中的课程，整个系列包含netty，xxl-job，rocketmq，nacos，sofajraft，spring，springboot，disruptor，编译器，虚拟机等等。
     * @author：陈清风扬，个人微信号：chenqingfengyangjj。
     * @方法描述：执行消息分发操作的方法
     */
    public void doDispatch(DispatchRequest req) throws RocksDBException {
        //在这里遍历所有的消息分发器
        for (CommitLogDispatcher dispatcher : this.dispatcherList) {
            //把消息分发请求交给每一个分发器处理
            dispatcher.dispatch(req);
        }
    }

    /**
     * @课程描述:从零带你写框架系列中的课程，整个系列包含netty，xxl-job，rocketmq，nacos，sofajraft，spring，springboot，disruptor，编译器，虚拟机等等。
     * @author：陈清风扬，个人微信号：chenqingfengyangjj。
     * @方法描述：消息重放操作执行完毕后的回调方法
     */
    public void finishCommitLogDispatch() {
        //该方法在源码中就是空实现
    }



    /**
     * @课程描述:从零带你写框架系列中的课程，整个系列包含netty，xxl-job，rocketmq，nacos，sofajraft，spring，springboot，disruptor，编译器，虚拟机等等。
     * @author：陈清风扬，个人微信号：chenqingfengyangjj。
     * @方法描述：根据消息物理偏移量获得对应的封装了消息内容的消息缓冲区
     */
    @Override
    public SelectMappedBufferResult selectOneMessageByOffset(long commitLogOffset) {
        //因为存储时编码的首4个字节写入的就是这条消息的长度，所以这里先把消息的长度取出来
        SelectMappedBufferResult sbr = this.commitLog.getMessage(commitLogOffset, 4);
        if (null != sbr) {
            try {
                //得到消息的实际长度
                int size = sbr.getByteBuffer().getInt();
                //根据消息长度得到真正的封装消息内容的缓冲区对象
                return this.commitLog.getMessage(commitLogOffset, size);
            } finally {
                //使用完毕之后释放缓冲区对象
                sbr.release();
            }
        }
        return null;
    }


    /**
     * @课程描述:从零带你写框架系列中的课程，整个系列包含netty，xxl-job，rocketmq，nacos，sofajraft，spring，springboot，disruptor，编译器，虚拟机等等。
     * @author：陈清风扬，个人微信号：chenqingfengyangjj。
     * @方法描述：得到CommitLog文件中消息的最小偏移量，其实就是commitLog文件队列中第一个文件的起始偏移量
     */
    @Override
    public long getMinPhyOffset() {
        return this.commitLog.getMinOffset();
    }




    /**
     * @课程描述:从零带你写框架系列中的课程，整个系列包含netty，xxl-job，rocketmq，nacos，sofajraft，spring，springboot，disruptor，编译器，虚拟机等等。
     * @author：陈清风扬，个人微信号：chenqingfengyangjj。
     * @方法描述：计算消息实际投递时间的方法，延迟消息会用到这个方法，在第十五版本代码，这个方法还用不到
     */
    public long computeDeliverTimestamp(final int delayLevel, final long storeTimestamp) {
        //从消息延迟级别对照表获取延迟级别对应的延迟时间
        Long time = this.delayLevelTable.get(delayLevel);
        if (time != null) {
            //实际投递时间就是存储时间加上延迟时间
            return time + storeTimestamp;
        }
        //如果没有找到对应的延迟时间，那就返回默认1秒延迟
        return storeTimestamp + 1000;
    }


    /**
     * @课程描述:从零带你写框架系列中的课程，整个系列包含netty，xxl-job，rocketmq，nacos，sofajraft，spring，springboot，disruptor，编译器，虚拟机等等。
     * @author：陈清风扬，个人微信号：chenqingfengyangjj。
     * @方法描述：得到消息最大延迟级别
     */
    public int getMaxDelayLevel() {
        return maxDelayLevel;
    }


    /**
     * @课程描述:从零带你写框架系列中的课程，整个系列包含netty，xxl-job，rocketmq，nacos，sofajraft，spring，springboot，disruptor，编译器，虚拟机等等。
     * @author：陈清风扬，个人微信号：chenqingfengyangjj。
     * @方法描述：得到消息确认偏移量的方法，在非集群模式下，这里返回的就是当前节点写入消息的最大偏移量
     */
    @Override
    public long getConfirmOffset() {
        return this.commitLog.getConfirmOffset();
    }



    /**
     * @课程描述:从零带你写框架系列中的课程，整个系列包含netty，xxl-job，rocketmq，nacos，sofajraft，spring，springboot，disruptor，编译器，虚拟机等等。
     * @author：陈清风扬，个人微信号：chenqingfengyangjj。
     * @方法描述：检查消息完整性的方法，在消息重放操作时，这个方法就会解析CommitLog文件中的一条条消息，得到要存储到消费队列中的消息
     */
    @Override
    public DispatchRequest checkMessageAndReturnSize(final ByteBuffer byteBuffer, final boolean checkCRC,
                                                     final boolean checkDupInfo, final boolean readBody) {
        return this.commitLog.checkMessageAndReturnSize(byteBuffer, checkCRC, checkDupInfo, readBody);
    }


    /**
     * @课程描述:从零带你写框架系列中的课程，整个系列包含netty，xxl-job，rocketmq，nacos，sofajraft，spring，springboot，disruptor，编译器，虚拟机等等。
     * @author：陈清风扬，个人微信号：chenqingfengyangjj。
     * @方法描述：当消费队列中有消息到达时，就意味着有消息可以消费者，当前方法就是通知相关组件有新消息到达的方法
     * 该方法在第十五版本代码中根本发挥不了作用，等后面用到了我才为大家真正讲解这个方法是怎么发挥作用的
     */
    @Override
    public void notifyMessageArriveIfNecessary(DispatchRequest dispatchRequest) {
        if (DefaultMessageStore.this.brokerConfig.isLongPollingEnable()
                && DefaultMessageStore.this.messageArrivingListener != null) {
            DefaultMessageStore.this.messageArrivingListener.arriving(dispatchRequest.getTopic(),
                    dispatchRequest.getQueueId(), dispatchRequest.getConsumeQueueOffset() + 1,
                    dispatchRequest.getTagsCode(), dispatchRequest.getStoreTimestamp(),
                    dispatchRequest.getBitMap(), dispatchRequest.getPropertiesMap());
            DefaultMessageStore.this.reputMessageService.notifyMessageArrive4MultiQueue(dispatchRequest);
        }
    }



    /**
     * @课程描述:从零带你写框架系列中的课程，整个系列包含netty，xxl-job，rocketmq，nacos，sofajraft，spring，springboot，disruptor，编译器，虚拟机等等。
     * @author：陈清风扬，个人微信号：chenqingfengyangjj。
     * @方法描述：消息重放服务组件，就是这个组件负责把CommitLog文件中的消息异步投递到消费队列和index文件队列中
     */
    class ReputMessageService extends ServiceThread {

        //消息重放的偏移量，这个偏移量其实就是记录当前组件从CommitLog的哪个位置开始处理消息的
        protected volatile long reputFromOffset = 0;

        public long getReputFromOffset() {
            return reputFromOffset;
        }

        public void setReputFromOffset(long reputFromOffset) {
            this.reputFromOffset = reputFromOffset;
        }


        /**
         * @课程描述:从零带你写框架系列中的课程，整个系列包含netty，xxl-job，rocketmq，nacos，sofajraft，spring，springboot，disruptor，编译器，虚拟机等等。
         * @author：陈清风扬，个人微信号：chenqingfengyangjj。
         * @方法描述：终止消息重放组件工作的方法
         */
        @Override
        public void shutdown() {
            //循环50次，每次都判断是否还有未重放的消息
            for (int i = 0; i < 50 && this.isCommitLogAvailable(); i++) {
                try {
                    //每次循环睡100ms，这里就是给够消息重放组件重放消息时间
                    //尽可能在停止工作之前把CommitLog中的消息重放完毕
                    Thread.sleep(100);
                } catch (InterruptedException ignored) {

                }
            }
            //判断组件终止之前是否仍有未重放的CommitLog消息
            if (this.isCommitLogAvailable()) {
                //如果有则记录告警日志
                LOGGER.warn("shutdown ReputMessageService, but CommitLog have not finish to be dispatched, CommitLog max" +
                                " offset={}, reputFromOffset={}", DefaultMessageStore.this.commitLog.getMaxOffset(),
                        this.reputFromOffset);
            }
            //调用父类关闭方法终止线程工作
            super.shutdown();
        }



        /**
         * @课程描述:从零带你写框架系列中的课程，整个系列包含netty，xxl-job，rocketmq，nacos，sofajraft，spring，springboot，disruptor，编译器，虚拟机等等。
         * @author：陈清风扬，个人微信号：chenqingfengyangjj。
         * @方法描述：计算还有多少CommitLog的数据没有重放成功的方法
         */
        public long behind() {
            return DefaultMessageStore.this.getConfirmOffset() - this.reputFromOffset;
        }


        /**
         * @课程描述:从零带你写框架系列中的课程，整个系列包含netty，xxl-job，rocketmq，nacos，sofajraft，spring，springboot，disruptor，编译器，虚拟机等等。
         * @author：陈清风扬，个人微信号：chenqingfengyangjj。
         * @方法描述：判断CommitLog文件中是否还有待重放的数据的方法
         */
        public boolean isCommitLogAvailable() {
            return this.reputFromOffset < DefaultMessageStore.this.getConfirmOffset();
        }



        /**
         * @课程描述:从零带你写框架系列中的课程，整个系列包含netty，xxl-job，rocketmq，nacos，sofajraft，spring，springboot，disruptor，编译器，虚拟机等等。
         * @author：陈清风扬，个人微信号：chenqingfengyangjj。
         * @方法描述：重放CommitLog中数据的方法，也是消息重放组件中最核心的方法
         */
        public void doReput() {
            //判断消息重放的偏移量是否小于CommitLog文件消息的最小偏移量，这里我要解释一下为什么要做这个判断
            //假如那么CommitLog文件的最小偏移量就是1000，而消息重放才重放到500，这就意味着CommitLog文件中前1000的数据已经被删除了
            //消息重放都没来得及重放，数据就被删除了，可能是消息重放的太慢，也可能是其他原因，导致CommitLog文件数据过期了消息重放还没处理完毕
            //所以这个时候要记录一下告警日志，然后更新消息重放的偏移量
            if (this.reputFromOffset < DefaultMessageStore.this.commitLog.getMinOffset()) {
                //在这里记录告警日志
                LOGGER.warn("The reputFromOffset={} is smaller than minPyOffset={}, this usually indicate that the dispatch behind too much and the commitlog has expired.",
                        this.reputFromOffset, DefaultMessageStore.this.commitLog.getMinOffset());
                //把消息重放偏移量设置为CommitLog文件最新的的最小偏移量
                this.reputFromOffset = DefaultMessageStore.this.commitLog.getMinOffset();
            }
            //在循环中执行消息重放操作，循环的时候一直判断CommitLog文件中是否还有待重放的数据
            //如果没有则退出循环
            for (boolean doNext = true; this.isCommitLogAvailable() && doNext; ) {
                //从要重访的位置得到封装了后面所有消息数据的缓冲区对象
                SelectMappedBufferResult result = DefaultMessageStore.this.commitLog.getData(reputFromOffset);
                //如果没有得到缓冲区对象，则直接退出循环
                if (result == null) {
                    break;
                }
                try {
                    //在循环中，每次循环都会更新消息重放起始位置为当前获取到的数据起始位置
                    this.reputFromOffset = result.getStartOffset();
                    //result.getSize()就是刚才从CommitLog文件中读取到的要重访的数据的总大小
                    //接下来要执行消息重放操作，所以这里会判断消息重放的大小如果超过刚才获得的数据总大小了，就会退出循环
                    for (int readSize = 0; readSize < result.getSize() && reputFromOffset < DefaultMessageStore.this.getConfirmOffset() && doNext; ) {
                        //这里会从CommitLog文件中真正解析消息，但是并不是解析消息的内容，而是得到消息在CommitLog中的全局偏移量，消息的大小，消息的目标队列Id，消息的tag过滤标签信息等等，这些信息都会封装到dispatchRequest对象中
                        DispatchRequest dispatchRequest = DefaultMessageStore.this.commitLog.checkMessageAndReturnSize(result.getByteBuffer(), false, false, false);
                        //得到本次解析到的消息的大小
                        int size = dispatchRequest.getBufferSize() == -1 ? dispatchRequest.getMsgSize() : dispatchRequest.getBufferSize();
                        //如果当前解析到的消息的大小加上消息重放位置，超过了CommitLog文件的消息确认位置，非集群模式下其实就是CommitLog文件的消息最大偏移量
                        if (reputFromOffset + size > DefaultMessageStore.this.getConfirmOffset()) {
                            //这个时候就会直接退出循环
                            doNext = false;
                            break;
                        }
                        //判断消息是否解析成功了
                        if (dispatchRequest.isSuccess()) {
                            if (size > 0) {
                                //如果消息解析成功了，那就可以在这里把消息交给消息重放器去重放
                                //这个时候消息就会被存储到消费队列和index文件队列中了
                                DefaultMessageStore.this.doDispatch(dispatchRequest);
                                //如果不是批量通知模式，则单条消息到达后，也就是被重放到消费队列中了，就直接通知相关组件处理
                                //这里的操作就是来一条消息就处理一条消息，不会攒一批了再通知
                                if (!notifyMessageArriveInBatch) {
                                    notifyMessageArriveIfNecessary(dispatchRequest);
                                }
                                //更新消息重放偏移量，增加当前重放的消息大小
                                this.reputFromOffset += size;
                                //更新已重放的消息大小
                                readSize += size;
                            }
                            //如果这次解析到的消息大小为0，说明本次得到的SelectMappedBufferResult缓冲区对象中的消息已经解析完毕了
                            //意味着需要切换到下一个文件开始解析了
                            else if (size == 0) {
                                //这里得到下一个内存映射文件的起始偏移量，把这个偏移量更新给消息重放偏移量
                                this.reputFromOffset = DefaultMessageStore.this.commitLog.rollNextFile(this.reputFromOffset);
                                //设置已读大小为当前SelectMappedBufferResult缓冲区数据大小，表示当前文件已处理完
                                readSize = result.getSize();
                            }
                        }
                        else {//执行到这里意味着消息没有解析成功
                            if (size > 0) {
                                //记录错误日志，日志信息为读取的消息总数与消息总大小不匹配
                                LOGGER.error("[BUG]read total count not equals msg total size. reputFromOffset={}", reputFromOffset);
                                //更新消息重放偏移量
                                this.reputFromOffset += size;
                            }
                            else {//下面代码逻辑大家暂时不必关心
                                doNext = false;
                                if (DefaultMessageStore.this.getMessageStoreConfig().isEnableDLegerCommitLog() ||
                                        DefaultMessageStore.this.brokerConfig.getBrokerId() == MixAll.MASTER_ID) {
                                    LOGGER.error("[BUG]dispatch message to consume queue error, COMMITLOG OFFSET: {}",
                                            this.reputFromOffset);
                                    this.reputFromOffset += result.getSize() - readSize;
                                }
                            }
                        }
                    }
                } catch (RocksDBException e) {
                    ERROR_LOG.info("dispatch message to cq exception. reputFromOffset: {}", this.reputFromOffset, e);
                    return;
                } finally {
                    //释放映射缓冲区资源
                    result.release();
                }
                //在消息重放操作执行完毕后，执行回调方法
                finishCommitLogDispatch();
            }
        }


        //该方法在第十五版本代码中用不到，我就先不添加注释了
        private void notifyMessageArrive4MultiQueue(DispatchRequest dispatchRequest) {
            Map<String, String> prop = dispatchRequest.getPropertiesMap();
            if (prop == null || dispatchRequest.getTopic().startsWith(MixAll.RETRY_GROUP_TOPIC_PREFIX)) {
                return;
            }
            String multiDispatchQueue = prop.get(MessageConst.PROPERTY_INNER_MULTI_DISPATCH);
            String multiQueueOffset = prop.get(MessageConst.PROPERTY_INNER_MULTI_QUEUE_OFFSET);
            if (StringUtils.isBlank(multiDispatchQueue) || StringUtils.isBlank(multiQueueOffset)) {
                return;
            }
            String[] queues = multiDispatchQueue.split(MixAll.MULTI_DISPATCH_QUEUE_SPLITTER);
            String[] queueOffsets = multiQueueOffset.split(MixAll.MULTI_DISPATCH_QUEUE_SPLITTER);
            if (queues.length != queueOffsets.length) {
                return;
            }
            for (int i = 0; i < queues.length; i++) {
                String queueName = queues[i];
                long queueOffset = Long.parseLong(queueOffsets[i]);
                int queueId = dispatchRequest.getQueueId();
                if (DefaultMessageStore.this.getMessageStoreConfig().isEnableLmq() && MixAll.isLmq(queueName)) {
                    queueId = 0;
                }
                DefaultMessageStore.this.messageArrivingListener.arriving(
                        queueName, queueId, queueOffset + 1, dispatchRequest.getTagsCode(),
                        dispatchRequest.getStoreTimestamp(), dispatchRequest.getBitMap(),
                        dispatchRequest.getPropertiesMap());
            }
        }


        /**
         * @课程描述:从零带你写框架系列中的课程，整个系列包含netty，xxl-job，rocketmq，nacos，sofajraft，spring，springboot，disruptor，编译器，虚拟机等等。
         * @author：陈清风扬，个人微信号：chenqingfengyangjj。
         * @方法描述：消息重放组件启动之后要运行的方法
         */
        @Override
        public void run() {
            DefaultMessageStore.LOGGER.info(this.getServiceName() + " service started");
            //循环执行操作
            while (!this.isStopped()) {
                try {
                    //每次让线程休息1毫秒，避免没有可重放数据时线程空转
                    TimeUnit.MILLISECONDS.sleep(1);
                    //执行消息重放操作
                    this.doReput();
                } catch (Exception e) {
                    DefaultMessageStore.LOGGER.warn(this.getServiceName() + " service has exception. ", e);
                }
            }
            DefaultMessageStore.LOGGER.info(this.getServiceName() + " service end");
        }


        /**
         * @课程描述:从零带你写框架系列中的课程，整个系列包含netty，xxl-job，rocketmq，nacos，sofajraft，spring，springboot，disruptor，编译器，虚拟机等等。
         * @author：陈清风扬，个人微信号：chenqingfengyangjj。
         * @方法描述：得到服务组件名称的方法
         */
        @Override
        public String getServiceName() {
            if (DefaultMessageStore.this.getBrokerConfig().isInBrokerContainer()) {
                return DefaultMessageStore.this.getBrokerIdentity().getIdentifier() + ReputMessageService.class.getSimpleName();
            }
            return ReputMessageService.class.getSimpleName();
        }

    }



    /**
     * @课程描述:从零带你写框架系列中的课程，整个系列包含netty，xxl-job，rocketmq，nacos，sofajraft，spring，springboot，disruptor，编译器，虚拟机等等。
     * @author：陈清风扬，个人微信号：chenqingfengyangjj。
     * @方法描述：消费队列数据刷盘器，这个刷新器会定期把消费队列的数据刷新到本地硬盘文件中
     */
    class FlushConsumeQueueService extends ServiceThread {

        //消息刷盘重试次数
        private static final int RETRY_TIMES_OVER = 3;

        //最新刷盘的时间
        private long lastFlushTimestamp = 0;


        /**
         * @课程描述:从零带你写框架系列中的课程，整个系列包含netty，xxl-job，rocketmq，nacos，sofajraft，spring，springboot，disruptor，编译器，虚拟机等等。
         * @author：陈清风扬，个人微信号：chenqingfengyangjj。
         * @方法描述：把消费队列数据刷盘的方法
         */
        private void doFlush(int retryTimes) {

            //得到消费队列每次刷盘数据的最少页数
            int flushConsumeQueueLeastPages = DefaultMessageStore.this.getMessageStoreConfig().getFlushConsumeQueueLeastPages();
            //判断消息刷盘重试次数是否到达上线
            if (retryTimes == RETRY_TIMES_OVER) {
                //到达上限则意味着要强制刷盘，所以这里把最小刷新页数设置为0，意味着不管有多少数据，都要立即执行刷盘操作
                flushConsumeQueueLeastPages = 0;
            }
            long logicsMsgTimestamp = 0;
            //得到消费队列数据刷新的时间间隔
            int flushConsumeQueueThoroughInterval = DefaultMessageStore.this.getMessageStoreConfig().getFlushConsumeQueueThoroughInterval();
            //得到当前系统时间
            long currentTimeMillis = System.currentTimeMillis();
            //判断距离上一次刷新的时间，是否超过了规定的时间间隔
            if (currentTimeMillis >= (this.lastFlushTimestamp + flushConsumeQueueThoroughInterval)) {
                //如果超过了则更新最新刷新时间
                this.lastFlushTimestamp = currentTimeMillis;
                //设置最小刷新页数为0，意味着可以强制执行刷盘操作
                flushConsumeQueueLeastPages = 0;
                //得到存储检查点中消费队列数据最新存储时间
                logicsMsgTimestamp = DefaultMessageStore.this.getStoreCheckpoint().getLogicsMsgTimestamp();
            }
            //得消费队列映射表，key是主题，value是队列Id和队列本身
            ConcurrentMap<String, ConcurrentMap<Integer, ConsumeQueueInterface>> tables = DefaultMessageStore.this.getConsumeQueueTable();
            //遍历队列执行刷新操作
            for (ConcurrentMap<Integer, ConsumeQueueInterface> maps : tables.values()) {
                //得到每一个消费队列
                for (ConsumeQueueInterface cq : maps.values()) {
                    //是否刷新成功的标志
                    boolean result = false;
                    //循环重试
                    for (int i = 0; i < retryTimes && !result; i++) {
                        //在这里执行刷盘操作
                        result = DefaultMessageStore.this.consumeQueueStore.flush(cq, flushConsumeQueueLeastPages);
                    }
                }
            }
            //如果是强制刷盘，则更新存储检查点中的逻辑消息时间戳
            if (0 == flushConsumeQueueLeastPages) {
                if (logicsMsgTimestamp > 0) {
                    DefaultMessageStore.this.getStoreCheckpoint().setLogicsMsgTimestamp(logicsMsgTimestamp);
                }
                //将存储检查数据刷盘
                DefaultMessageStore.this.getStoreCheckpoint().flush();
            }
        }



        /**
         * @课程描述:从零带你写框架系列中的课程，整个系列包含netty，xxl-job，rocketmq，nacos，sofajraft，spring，springboot，disruptor，编译器，虚拟机等等。
         * @author：陈清风扬，个人微信号：chenqingfengyangjj。
         * @方法描述：消费队列刷盘器启动后要执行的方法
         */
        @Override
        public void run() {
            DefaultMessageStore.LOGGER.info(this.getServiceName() + " service started");
            while (!this.isStopped()) {
                try {
                    //得到消费队列数据刷盘间隔
                    int interval = DefaultMessageStore.this.getMessageStoreConfig().getFlushIntervalConsumeQueue();
                    this.waitForRunning(interval);
                    //执行刷盘操作
                    this.doFlush(1);
                } catch (Exception e) {
                    DefaultMessageStore.LOGGER.warn(this.getServiceName() + " service has exception. ", e);
                }
            }//程序关闭之前再执行一次刷盘操作
            this.doFlush(RETRY_TIMES_OVER);
            DefaultMessageStore.LOGGER.info(this.getServiceName() + " service end");
        }

        /**
         * @课程描述:从零带你写框架系列中的课程，整个系列包含netty，xxl-job，rocketmq，nacos，sofajraft，spring，springboot，disruptor，编译器，虚拟机等等。
         * @author：陈清风扬，个人微信号：chenqingfengyangjj。
         * @方法描述：得到服务名称的方法
         */
        @Override
        public String getServiceName() {
            if (DefaultMessageStore.this.brokerConfig.isInBrokerContainer()) {
                return DefaultMessageStore.this.getBrokerIdentity().getIdentifier() + FlushConsumeQueueService.class.getSimpleName();
            }
            return FlushConsumeQueueService.class.getSimpleName();
        }


        //等待线程结束工作的最大时间
        @Override
        public long getJoinTime() {
            return 1000 * 60;
        }
    }




    /**
     * @课程描述:从零带你写框架系列中的课程，整个系列包含netty，xxl-job，rocketmq，nacos，sofajraft，spring，springboot，disruptor，编译器，虚拟机等等。
     * @author：陈清风扬，个人微信号：chenqingfengyangjj。
     * @方法描述：ConsumeQueue消费队列消息分发器，这个分发器会把CommitLog文件中的消息分发到ConsumeQueue文件中去
     *注意，这里我要解释一下，分发的并不是消息内容本身，而是消息的全局偏移量，消息长度，消息过滤标签信息
     */
    class CommitLogDispatcherBuildConsumeQueue implements CommitLogDispatcher {



        /**
         * @课程描述:从零带你写框架系列中的课程，整个系列包含netty，xxl-job，rocketmq，nacos，sofajraft，spring，springboot，disruptor，编译器，虚拟机等等。
         * @author：陈清风扬，个人微信号：chenqingfengyangjj。
         * @方法描述：处理消息分发请求的方法
         */
        @Override
        public void dispatch(DispatchRequest request) throws RocksDBException {
            //得到消息事务类型
            final int tranType = MessageSysFlag.getTransactionValue(request.getSysFlag());
            switch (tranType) {
                //如果是已提交的事务消息和非事务消息
                case MessageSysFlag.TRANSACTION_NOT_TYPE:
                case MessageSysFlag.TRANSACTION_COMMIT_TYPE:
                    //那就在这个方法中执行消息分发操作，也就是把分发消息请求交给consumeQueueStore消费队列存储器处理
                    putMessagePositionInfo(request);
                    break;
                case MessageSysFlag.TRANSACTION_PREPARED_TYPE:
                case MessageSysFlag.TRANSACTION_ROLLBACK_TYPE:
                    break;
            }
        }
    }



    /**
     * @课程描述:从零带你写框架系列中的课程，整个系列包含netty，xxl-job，rocketmq，nacos，sofajraft，spring，springboot，disruptor，编译器，虚拟机等等。
     * @author：陈清风扬，个人微信号：chenqingfengyangjj。
     * @方法描述：消费队列存储器存放被分发的消息的方法
     */
    protected void putMessagePositionInfo(DispatchRequest dispatchRequest) throws RocksDBException {
        this.consumeQueueStore.putMessagePositionInfoWrapper(dispatchRequest);
    }


    /**
     * @课程描述:从零带你写框架系列中的课程，整个系列包含netty，xxl-job，rocketmq，nacos，sofajraft，spring，springboot，disruptor，编译器，虚拟机等等。
     * @author：陈清风扬，个人微信号：chenqingfengyangjj。
     * @方法描述：这个消息分发器是用来分发index文件消息的，只不过在第十五版本代码中并没有引入index文件，所以我先把这个分发器注释了
     */
    class CommitLogDispatcherBuildIndex implements CommitLogDispatcher {

        @Override
        public void dispatch(DispatchRequest request) {
            if (DefaultMessageStore.this.messageStoreConfig.isMessageIndexEnable()) {
                DefaultMessageStore.this.indexService.buildIndex(request);
            }
        }
    }
}
